/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package service

import (
	"context"
	"net"
	"net/rpc"
	"net/rpc/jsonrpc"

	"hundsun.com/hsl/hschain/blockchain/impl/pubsub"

	"hundsun.com/hsl/hschain/common/config"
	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/common/types"
	pbcom "hundsun.com/hsl/hschain/protos/common"
	protosrpc "hundsun.com/hsl/hschain/protos/rpc"
	"hundsun.com/hsl/hschain/rpc/common"
	"hundsun.com/hsl/hschain/store"
	dbm "hundsun.com/hsl/hschain/store/db"
	"hundsun.com/hsl/hschain/store/mq"
)

//IRpcMethod IRpcMethod接口
type IRpcMethod interface {
	GetSubMethodAndParam(string) (interface{}, interface{}, error)
	GetRpcMethodAndParam(string) (interface{}, interface{}, error)
}

//RPCService RPCService
type RPCService struct {
	config              *config.ModuleConfig
	rpcConfig           *common.RPCConfig
	rpcServer           *server
	Log                 logging.Log
	workerClient        mq.Client
	topicClient         mq.Client
	rpcDb               dbm.Database
	ctx                 context.Context
	module              *rpcModule
	errChan             chan<- GoroutineErrorWrapper
	endpointInstance    *EndPointMonitor
	flowControlInstance *FlowControlMonitor
}

//Init 初始化
func (p *RPCService) Init(ctx context.Context, module *rpcModule, errChan chan<- GoroutineErrorWrapper) error {
	p.Log = module.Log
	p.ctx = ctx
	p.module = module
	p.config = module.Config
	p.errChan = errChan
	p.rpcDb = store.GetRPCDB()
	p.rpcConfig = common.NewRPCConfig(p.config.Parameters)
	p.flowControlInstance = NewFlowControlInstance(ctx, p.rpcConfig.GetTpsLimit(), module.Log)
	wClient, err := store.NewMqClient(mq.ModeWorker.String())
	if err != nil {
		panic(err)
	}
	wClient.Sub(types.RPCModule, mq.ModeWorker.String())
	p.workerClient = wClient

	tClient, err := store.NewMqClient(mq.ModeTopic.String())
	if err != nil {
		panic(err)
	}
	tClient.Sub(types.RPCModule, mq.ModeTopic.String())
	p.topicClient = tClient
	p.endpointInstance = NewEndPointMonitor(ctx, p.Log, wClient, p.rpcDb)
	return nil
}

//关闭rpc服务端
func (p *RPCService) closeRPCServer() error {
	if p.rpcServer == nil {
		return nil
	}
	p.rpcServer.grpcServer.Stop()
	return nil
}

//启动rpc服务端
func (p *RPCService) startRPCServer(grpcPort string, JSONRPCPort string) error {
	p.rpcServer = newRPCServerInstance(p)

	//json rpc服务
	go func() {
		rpc.NewServer()
		err := rpc.Register(JSONRPCServer{p.rpcServer})
		if err != nil {
			panic(err)
		}
		listener, err := net.Listen("tcp", "0.0.0.0:"+JSONRPCPort)
		if err != nil {
			panic(err)
		}
		for {
			// 不断连接服务
			select {
			case <-p.ctx.Done():
				p.Log.Infof("json rpc server close, port:%s", JSONRPCPort)
				listener.Close()
				return
			default:
				conn, err := listener.Accept()
				if err != nil {
					p.Log.Infof("json rpc server started, port:%s", JSONRPCPort)
					continue
				}
				// 使用 Goroutine：ServeConn runs the JSON-RPC server on a single connection.
				go jsonrpc.ServeConn(conn)
			}
		}
	}()

	//grpc服务
	go func() {
		grpcListener, err := net.Listen("tcp", "0.0.0.0:"+grpcPort)
		if err != nil {
			panic(err)
		}

		protosrpc.RegisterRpcServer(p.rpcServer.grpcServer, p.rpcServer)
		err = p.rpcServer.grpcServer.Serve(grpcListener)
		if err != nil {
			panic(err)
		}
	}()
	//订阅blockHeader
	err := p.subMsg(&pbcom.Router{
		MainTopic: types.RPCModule,
		SubTopic:  mq.ModeTopic.String(),
		EventType: pubsub.SubBlockHeader,
	})
	if err != nil {
		return err
	}
	p.Log.Infof("rpc server started, port:%s", grpcPort)
	return nil
}

func (p *RPCService) subMsg(router *pbcom.Router) error {
	//封装订阅消息
	data := &pbcom.ReqSubscriber{
		MsgRouter: router,
		Type:      pubsub.SubBlockHeader,
	}
	msg := p.workerClient.NewMessage(types.BlockChainModule, mq.ModeWorker.String(), types.EventSubscriber, data)
	err := p.workerClient.Send(msg, false)
	if err != nil {
		return err
	}
	return nil
}

func (p *RPCService) procTopicMqMsg() {
	p.Log.Infof("rpc subscribe started")

	for msg := range p.topicClient.Recv() {
		if msg == nil {
			continue
		}
		switch msg.Ty {
		case pubsub.SubBlockHeader:
			go mq.ProcessMsg(msg, p.endpointInstance.Consume, nil, p.Log)
		}
	}
}
